package com.rxjava.simpleio;

/**
 * 被观察者，用来生产发送事件；
 */
public class Observable<T> {

    private OnSubscribe<T> onSubscribe;

    private Observable(OnSubscribe<T> onSubscribe) {
    	//onSubscribe -> subscriber.onNext(3);
    	//1map -> onSubscribe -> OnSubscribeLift对象;
    	//thread -> onSubscribe -> OnSubscribeOnIO对象;
        this.onSubscribe = onSubscribe;
    }

    public final void subscribe(Subscriber<? super T> subscriber) {
    	System.out.println("step1");
    	onSubscribe.call(subscriber);
    }

    public static <T> Observable<T> create(OnSubscribe<T> onSubscribe) {
    	//onSubscribe -> subscriber.onNext(3);
    	//thread -> onSubscribe -> OnSubscribeOnIO对象;
        return new Observable<>(onSubscribe);
    }

    public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
    	//1map -> func -> return "map" + integer;
        return lift(new OperatorMap<T, R>(func));
    }

    public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
    	//1map -> onSubscribe -> subscriber.onNext(3);
    	//1map -> operator -> OperatorMap对象
        return new Observable<>(new OnSubscribeLift<T, R>(onSubscribe, operator));
    }

    public final Observable<T> subscribeOnIO() {
    	//thread -> this -> Observable对象(OnSubscribeLift对象)
        return create(new OnSubscribeOnIO<T>(this));
    }

}